package com.bkhech.zk;

import com.bkhech.pojo.netty.NettyServerNode;
import com.bkhech.utils.JsonUtils;
import com.bkhech.utils.RedisOperator;
import jakarta.annotation.Resource;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Objects;

import static com.bkhech.RedisKeyConstant.NETTY_PORT_KEY;
import static com.bkhech.RedisKeyConstant.NETTY_QUEUE_NAME_PREFIX;
import static com.bkhech.ZkConstant.NODE_NAME;

@Slf4j
@Configuration
@Data
@ConfigurationProperties(prefix = "zookeeper.curator")
public class CuratorConfig {

    @Resource
    private RedisOperator redisOperator;

    @Resource
    private RabbitAdmin rabbitAdmin;

    // 单机/集群，格式 ip:port,ip2:port2
    private String host;
    // 连接超时时间
    private Integer connectionTimeoutMs;
    // 会话连接超时时间
    private Integer sessionTimeoutMs;
    // 每次重试的间隔时间
    private Integer sleepMsBetweenRetry;
    // 最大重试次数
    private Integer maxRetries;
    // 命名空间（root 根节点名称）
    private String namespace;

    @Bean(value = "curatorClient")
    public CuratorFramework curatorClient() {
        // 定义重试策略
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(sleepMsBetweenRetry, maxRetries);
        // 声明初始化客户端
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .retryPolicy(retryPolicy)
                .connectionTimeoutMs(connectionTimeoutMs)
                .sessionTimeoutMs(sessionTimeoutMs)
                .namespace(namespace)
                .connectString(host)
                .build();
        // 启动 curator 客户端
        client.start();
        log.info("CuratorFramework 客户端已经成功启动！！！！");
        // 注册监听事件
        addWatcher(NODE_NAME, client);
        return client;
    }

    /**
     * 注册节点的事件监听
     *
     * @param path   监听的节点
     * @param client 客户端
     */
    public void addWatcher(String path, CuratorFramework client) {
        CuratorCache curatorCache = CuratorCache.build(client, path);
        curatorCache.listenable().addListener((type, oldData, data) -> {
            log.info("节点更新后的数据、状态: {}", data);

            final CuratorCacheListener.Type eventType = type;
            switch (eventType) {
                case NODE_CREATED -> log.info("子节点创建事件");
                case NODE_CHANGED -> log.info("子节点更新事件");
                case NODE_DELETED -> {
                    log.info("子节点删除事件");
                    String oldDataJson = new String(oldData.getData());
                    NettyServerNode oldNode = JsonUtils.jsonToPojo(oldDataJson, NettyServerNode.class);
                    log.info("old path: {}, old node: {}", oldData.getPath(), oldNode);
                    if (Objects.nonNull(oldNode)) {
                        String oldPort = oldNode.getPort() + "";
                        String portKey = NETTY_PORT_KEY;

                        // 删除残留缓存端口
                        redisOperator.hdel(portKey, oldPort);

                        // 删除mq中残留的节点
                        String queueName = NETTY_QUEUE_NAME_PREFIX + oldPort;
                        rabbitAdmin.deleteQueue(queueName);
                    }
                }
            }
        });

        curatorCache.start();
    }


}
